package com.atguigu.edu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.app.func.BeanToJsonStrFunction;
import com.atguigu.edu.realtime.app.func.DimAsyncFunction;
import com.atguigu.edu.realtime.bean.TradeCourseOrderBean;
import com.atguigu.edu.realtime.util.DateFormatUtil;
import com.atguigu.edu.realtime.util.DorisUtil;
import com.atguigu.edu.realtime.util.MyKafkaUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.util.concurrent.TimeUnit;

/**
 * ClassName: DwsTradeCourseOrderWindow
 * Package: com.atguigu.edu.realtime.app.dws
 * Description:
 * 课程粒度下单业务过程聚合统计
 * @Author Mr.2
 * @Create 2023/9/11 18:51
 * @Version 1.0
 */
public class DwsTradeCourseOrderWindow {
    public static void main(String[] args) {
        // TODO 1. 基本环境准备
        // 1.1 创建流式执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 1.2 设置并行度
        env.setParallelism(4);

        // TODO 2. 检查点相关设置
        // 2.1 开启检查点; 往Doris写必须开启检查点
        env.enableCheckpointing(10000L, CheckpointingMode.EXACTLY_ONCE);
        // 2.2 设置 检查点超时时间
        // 2.3 设置 job取消后, 检查点是否保留
        // 2.4 设置 检查点重启策略
        // 2.5 设置 两个检查点最小时间间隔
        // 2.6 设置 状态后端
        // 2.7 设置系统操作用户 hadoop_user_name

        // TODO 3. From kafka topic 读取数据
        // 3.1 声明 消费者主题 和 消费者组
        String topic = "dwd_trade_order_detail";
        String groupId = "dws_trade_course_order_group_01";
        // 3.2 创建消费者对象
        KafkaSource<String> kafkaSource = MyKafkaUtil.getKafkaSource(topic, groupId);
        // 3.3 消费数据 封装为流
        DataStreamSource<String> kafkaStrStream =
                env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka-source");

        // TODO 4. 过滤空消息, 数据类型转换 JSONString -> JSONObject
        SingleOutputStreamOperator<JSONObject> jsonObjStream = kafkaStrStream.process(
                new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String jsonStr, Context ctx, Collector<JSONObject> out) throws Exception {
                        if (StringUtils.isNotEmpty(jsonStr)) {
                            JSONObject jsonObject = JSON.parseObject(jsonStr);
                            out.collect(jsonObject);
                        }
                    }
                }
        );

        // TODO 5. 去重
        KeyedStream<JSONObject, String> orderDetailIdKeyedStream
                = jsonObjStream.keyBy(jsonObj -> jsonObj.getString("id"));
        // 5.1 去重方式1: 状态 + 定时任务
        // 5.2 推荐使用 --> 去重方式2: 状态 + 抵消
        SingleOutputStreamOperator<JSONObject> distinctStream = orderDetailIdKeyedStream.process(
                new KeyedProcessFunction<String, JSONObject, JSONObject>() {

                    // 声明状态
                    private ValueState<JSONObject> lastJsonObjState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        // 状态初始化
                        ValueStateDescriptor<JSONObject> jsonObjValueStateDescriptor =
                                new ValueStateDescriptor<>("lastJsonObjState", JSONObject.class);
                        // 思考: 对状态保留时间的设置 TTL 10s
                        jsonObjValueStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.seconds(10L)).build());
                        lastJsonObjState = getRuntimeContext().getState(jsonObjValueStateDescriptor);
                    }

                    @Override
                    public void processElement(JSONObject jsonObj, Context ctx, Collector<JSONObject> out) throws Exception {
                        // 获取上次状态
                        JSONObject lastJsonObj = lastJsonObjState.value();

                        if (lastJsonObj != null) {
                            String originAmount = lastJsonObj.getString("origin_amount");
                            String couponReduce = lastJsonObj.getString("coupon_reduce");
                            String finalAmount = lastJsonObj.getString("final_amount");

                            // 取反 去重
                            lastJsonObj.put("origin_amount", "-" + originAmount);
                            lastJsonObj.put("coupon_reduce", "-" + couponReduce);
                            lastJsonObj.put("final_amount", "-" + finalAmount);

                            // 将取反的数据 往下游传递
                            out.collect(lastJsonObj);
                        }

                        // 第1条数据， 直接往下游传递
                        out.collect(jsonObj);
                        // 更新到状态中
                        lastJsonObjState.update(jsonObj);
                    }
                }
        );

        // TODO 6. 对流中数据进行类型转换, JSONObject -> 实体类对象(TradeCourseOrderBean)
        SingleOutputStreamOperator<TradeCourseOrderBean> orderBeanStream = distinctStream.map(
                new MapFunction<JSONObject, TradeCourseOrderBean>() {
                    @Override
                    public TradeCourseOrderBean map(JSONObject jsonObj) throws Exception {

                        String courseId = jsonObj.getString("course_id");

                        String originAmount = jsonObj.getString("origin_amount");
                        String couponReduce = jsonObj.getString("coupon_reduce");
                        String finalAmount = jsonObj.getString("final_amount");
                        // 时间单位: s -> ms
                        Long ts = jsonObj.getLong("ts") * 1000L;

                        // 使用 构造者模式; 需要对 jsonObject 的 Null值 作处理, 转成对应的数值0.0 。。。
                        TradeCourseOrderBean courseBean = TradeCourseOrderBean.builder()
                                .courseId(courseId)
                                .originAmount(new BigDecimal(originAmount == null ? "0.0" : originAmount))
                                .couponReduce(new BigDecimal(couponReduce == null ? "0.0" : couponReduce))
                                .finalAmount(new BigDecimal(finalAmount == null ? "0.0" : finalAmount))
                                .ts(ts)
                                .build();

                        return courseBean;
                    }
                }
        );
        // For test output
//        orderBeanStream.print("bean->");

        // TODO 7. 设置Watermark 以及 提取事件时间(event time)字段
        SingleOutputStreamOperator<TradeCourseOrderBean> withWatermarkStream = orderBeanStream.assignTimestampsAndWatermarks(
                WatermarkStrategy.<TradeCourseOrderBean>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<TradeCourseOrderBean>() {
                                    @Override
                                    public long extractTimestamp(TradeCourseOrderBean orderBean, long recordTimestamp) {
                                        return orderBean.getTs();
                                    }
                                }
                        )
        );

        // TODO 8. 按照维度course_id进行分组
        KeyedStream<TradeCourseOrderBean, String> courseIdKeyedStream =
                withWatermarkStream.keyBy(TradeCourseOrderBean::getCourseId);

        // TODO 9. 开窗
        WindowedStream<TradeCourseOrderBean, String, TimeWindow> windowedStream =
                courseIdKeyedStream.window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10L)));

        // TODO 10. 聚合计算
        SingleOutputStreamOperator<TradeCourseOrderBean> reducedStream = windowedStream.reduce(
                new ReduceFunction<TradeCourseOrderBean>() {
                    @Override
                    public TradeCourseOrderBean reduce(TradeCourseOrderBean value1, TradeCourseOrderBean value2) throws Exception {
                        value1.setOriginAmount(value1.getOriginAmount().add(value2.getOriginAmount()));
                        value1.setCouponReduce(value1.getCouponReduce().add(value2.getCouponReduce()));
                        value1.setFinalAmount(value1.getFinalAmount().add(value2.getFinalAmount()));
                        return value1;
                    }
                },
                new WindowFunction<TradeCourseOrderBean, TradeCourseOrderBean, String, TimeWindow>() {
                    @Override
                    public void apply(String key, TimeWindow window, Iterable<TradeCourseOrderBean> input, Collector<TradeCourseOrderBean> out) throws Exception {
                        String stt = DateFormatUtil.toYmdHms(window.getStart());
                        String edt = DateFormatUtil.toYmdHms(window.getEnd());
                        String curDate = DateFormatUtil.toDate(window.getStart());

                        for (TradeCourseOrderBean orderBean : input) {
                            orderBean.setStt(stt);
                            orderBean.setEdt(edt);
                            orderBean.setCurDate(curDate);
                            // 向下游传递
                            out.collect(orderBean);
                        }
                    }
                }
        );
        // For test output -> yes
//        reducedStream.print("reduced->");

        // TODO 12. 关联维度 course_id
        SingleOutputStreamOperator<TradeCourseOrderBean> withCourseNameStream = AsyncDataStream.unorderedWait(
                reducedStream,
                new DimAsyncFunction<TradeCourseOrderBean>("dim_course_info") {
                    @Override
                    public String getKey(TradeCourseOrderBean courseOrderBean) {
                        return courseOrderBean.getCourseId();
                    }

                    @Override
                    public void join(TradeCourseOrderBean courseOrderBean, JSONObject dimInfoJsonObj) {
                        courseOrderBean.setCourseName(dimInfoJsonObj.getString("course_name"));
                        courseOrderBean.setSubjectId(dimInfoJsonObj.getString("subject_id"));
                    }
                },
                60L,
                TimeUnit.SECONDS
        );
        // TODO 13. 关联维度 subject_name
        SingleOutputStreamOperator<TradeCourseOrderBean> withSubjectNameStream = AsyncDataStream.unorderedWait(
                withCourseNameStream,
                new DimAsyncFunction<TradeCourseOrderBean>("dim_base_subject_info") {
                    @Override
                    public String getKey(TradeCourseOrderBean courseOrderBean) {
                        return courseOrderBean.getSubjectId();
                    }

                    @Override
                    public void join(TradeCourseOrderBean courseOrderBean, JSONObject dimInfoJsonObj) {
                        courseOrderBean.setSubjectName(dimInfoJsonObj.getString("subject_name"));
                        courseOrderBean.setCategoryId(dimInfoJsonObj.getString("category_id"));
                    }
                },
                60L,
                TimeUnit.SECONDS
        );
        // TODO 14. 关联维度 category_name
        SingleOutputStreamOperator<TradeCourseOrderBean> resultStream = AsyncDataStream.unorderedWait(
                withSubjectNameStream,
                new DimAsyncFunction<TradeCourseOrderBean>("dim_base_category_info") {
                    @Override
                    public String getKey(TradeCourseOrderBean courseOrderBean) {
                        return courseOrderBean.getCategoryId();
                    }

                    @Override
                    public void join(TradeCourseOrderBean courseOrderBean, JSONObject dimInfoJsonObj) {
                        courseOrderBean.setCategoryName(dimInfoJsonObj.getString("category_name"));
                    }
                },
                60L,
                TimeUnit.SECONDS
        );
        // For test output
        resultStream.print(">>>>>");

        // TODO 11. 写到Doris
        resultStream
                .map(new BeanToJsonStrFunction<TradeCourseOrderBean>())
                .sinkTo(DorisUtil.getDorisSink("dws_trade_course_order_window"));


        // 执行环境
        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

    }
}
